import asyncio
import json
import multiprocessing
import time
import logging
import hashlib
import threading
import urllib.parse

import pandas
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
from django.db import close_old_connections, IntegrityError

from paramiko.client import SSHClient, AutoAddPolicy

from django.db.models import Q
from django.http import JsonResponse
from rest_framework.views import APIView

from hosts_mgr.models import GroupList, HostList, GroupHost, CmdRecord, FileDistribution
from role.models import SysUserRole, SysRoleGroupList

from utils.encrypt import Password
from utils.page_data import PageData
from utils.ssh_client import LocalSSHClient
from utils.file_response import FileResponseLocal
from utils.check_permission import check_permission

logger = logging.getLogger()
# 修改权限
logger.setLevel(logging.INFO)


class WithOpenFile:
    """打开文件并发送到服务器"""

    def open_file(self, file_obj, sftp, remote_path, group_name, total_size=0):
        if not total_size:
            total_size = file_obj.size
        uploaded_size = 0
        # 在循环外初始化变量
        last_progress = 0
        last_send_time = time.time()
        progress_interval = 5  # 进度变化阈值5%
        time_interval = 1  # 时间间隔1秒
        # 获取Channel层
        channel_layer = get_channel_layer()
        # 上传文件
        with sftp.file(remote_path, 'wb') as f:
            while True:
                chunk = file_obj.read(4096)
                if not chunk:
                    break
                f.write(chunk)

                uploaded_size += len(chunk)
                current_progress = int((uploaded_size / total_size) * 100)
                # 计算上传进度
                # 发送进度信息
                # 向指定组发送消息（假设组名为 terminal）
                # 当进度变化超过阈值时发送消息
                # 双重触发条件：进度变化≥5% 或 时间间隔≥1秒
                if (current_progress - last_progress >= progress_interval) or \
                        (time.time() - last_send_time >= time_interval):
                    async_to_sync(channel_layer.group_send)(
                        group_name,  # 组名需与 WebSocket 路由中的 group_name 一致
                        {
                            "type": "send.message",  # 对应 SSHConsumer 中的方法名（下划线式命名）
                            "message": {
                                "code": 200,
                                "type": "progress",
                                "data": current_progress  # 发送实际进度而非分块进度
                            }
                        }
                    )
                    last_progress = current_progress
                    last_send_time = time.time()
            # 强制发送最终进度
            if last_progress < 100:
                async_to_sync(channel_layer.group_send)(
                    group_name,  # 组名需与 WebSocket 路由中的 group_name 一致
                    {
                        "type": "send.message",  # 对应 SSHConsumer 中的方法名（下划线式命名）
                        "message": {
                            "code": 200,
                            "type": "progress",
                            "data": 100  # 发送实际进度而非分块进度
                        }
                    }
                )


class FileSrcToDestDistributionView(APIView):
    """源服务器到目标服务器文件分发"""

    @check_permission(perms_list=['file:list'])  # 直接应用装饰器，无需method_decorator
    def dispatch(self, request, *args, **kwargs):
        return super().dispatch(request, *args, **kwargs)

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.__encryption_pass = Password()

    def post(self, request):
        data = json.loads(request.body.decode("utf-8"))
        dest_host_list = data["dest_host_list"]
        src_host_id = data["src_host_id"]
        dest_path = data["dest_path"].strip()
        src_path = data["src_path"].strip()

        threads = []
        errors = []
        lock = threading.Lock()

        src_host_obj = HostList.objects.filter(id=src_host_id).first()
        # 解密密码
        password = self.__encryption_pass.decrypt(src_host_obj.password)

        # 创建并启动线程
        for dest_host_id in dest_host_list:
            dest_host_obj = HostList.objects.filter(id=dest_host_id).first()

            if src_host_obj.server_type == 2:
                password = self.__encryption_pass.decrypt(dest_host_obj.password)

            thread = threading.Thread(target=self.__file_thread,
                                      args=(request, dest_path, password, src_path, lock, errors, src_host_obj,
                                            dest_host_obj))
            threads.append(thread)
            thread.start()

        # 等待所有线程完成
        for thread in threads:
            thread.join()

        if errors:
            return JsonResponse({
                "code": 207,
                "msg": f"文件分发完成，失败{len(errors)}台,errors: {errors}",
            })

        return JsonResponse({"code": 200, "msg": "文件全部分发成功"})

    def __file_thread(self, request, dest_path, password, src_path, lock, errors, src_host_obj,
                      dest_host_obj):
        file_record = FileDistributionView()
        user = request.session.get('username')
        path = f"{src_path} => {dest_path}"
        try:
            if src_host_obj.server_type == 2 and dest_host_obj.server_type == 1:
                client = LocalSSHClient(ip_addr=dest_host_obj.connect_ip, port=dest_host_obj.port,
                                        username=dest_host_obj.user, password=password)
                client.execute_command_progress(dest_path, dest_host_obj, src_path, src_host_obj)
            elif src_host_obj.server_type == 1:
                client = LocalSSHClient(ip_addr=src_host_obj.connect_ip, port=src_host_obj.port,
                                        username=src_host_obj.user, password=password)
                client.execute_command_progress(dest_path, dest_host_obj, src_path, src_host_obj)
            elif src_host_obj.server_type == 2 and dest_host_obj.server_type == 2:
                get_sftp = TerminalFileView()
                dest_client, ssh_client = get_sftp.get_sftp_connection(dest_host_obj.id)

                # 获取前端ws连接过来的 file_name + host_id 的md5值
                md5 = hashlib.md5()
                md5.update(f"{dest_path}{dest_host_obj.id}".encode('utf-8'))
                md5_group = md5.hexdigest()

                get_sftp = TerminalFileView()
                src_sftp_client = get_sftp.get_sftp_connection(src_host_obj.id)[0]
                # 获取文件属性以确定大小
                file_attr = src_sftp_client.stat(src_path)
                file_size = file_attr.st_size

                file_obj = src_sftp_client.open(src_path, 'rb')

                # 初始化文件传输类
                with_open_file = WithOpenFile()
                with_open_file.open_file(file_obj=file_obj, sftp=dest_client,
                                         remote_path=dest_path, group_name=md5_group, total_size=file_size)

            logger.info(f"目标主机：{dest_host_obj.hostname}传输完毕")
            file_record.file_record(path, username=user, code=1, dest_hostname=dest_host_obj.hostname,
                                    src_hostname=src_host_obj.hostname)
        except Exception as e:
            error_msg = f"主机 {src_host_obj.hostname} 分发失败: {str(e)}"
            file_record.file_record(path, username=user, code=2, dest_hostname=dest_host_obj.hostname,
                                    src_hostname=src_host_obj.hostname)
            with lock:
                errors.append(error_msg)
            logger.error(error_msg)


class FileDistributionView(APIView):
    """本地文件分发到目标服务器功能类"""

    @check_permission(perms_list=['file:list'])  # 直接应用装饰器，无需method_decorator
    def dispatch(self, request, *args, **kwargs):
        return super().dispatch(request, *args, **kwargs)

    def get(self, request):
        user = request.session.get('username')
        file_distribution_list = (FileDistribution.objects.filter(user=user)
                                  .order_by('-id').values("path", "hostname", "user",
                                                          "status", "distribution_time"))
        for item in file_distribution_list:
            try:
                item['distribution_time'] = item['distribution_time'].strftime('%Y-%m-%d %H:%M:%S')
            except AttributeError as e:
                logger.error(str(e))
        return JsonResponse({"code": 200, "msg": list(file_distribution_list),
                             "total": file_distribution_list.count()})

    def post(self, request):
        """文件上传"""
        file_obj = request.FILES.get('files')
        data = request.POST.dict()

        file_name = file_obj.name
        path = data.get("path").rstrip('/')

        # 新增：前端传来 host_list 的分发请求
        host_list = data.get("host_list")
        if host_list:
            host_ids = host_list.split(',')
            threads = []
            errors = []
            lock = threading.Lock()

            username = request.session.get('username')
            # 创建并启动线程
            for host_id in host_ids:
                host_obj = HostList.objects.filter(id=host_id).get()
                thread = threading.Thread(target=self.__upload_thread,
                                          args=(host_id, host_obj, file_obj, path,
                                                file_name, lock, errors, username))
                threads.append(thread)
                thread.start()

            # 等待所有线程完成
            for thread in threads:
                thread.join()

            if errors:
                return JsonResponse({
                    "code": 207,
                    "msg": f"文件分发完成，失败{len(errors)}台,errors: {errors}",
                })
            return JsonResponse({"code": 200, "msg": "文件全部分发成功"})

    def __upload_thread(self, host_id, host_obj, file_obj, path, file_name, lock, errors, username):
        sftp = None
        ssh_client = None
        error_msg = ''
        try:
            get_sftp = TerminalFileView()
            sftp, ssh_client = get_sftp.get_sftp_connection(host_id)

            if host_obj.server_type == 1:
                remote_path = f"{path}/{file_name}"
            else:
                remote_path = fr"{path}\{file_name}"
            # 获取前端ws连接过来的 file_name + host_id 的md5值
            md5 = hashlib.md5()
            md5.update(f"{file_name}{host_id}".encode('utf-8'))
            md5_group = md5.hexdigest()

            # 初始化文件传输类
            with_open_file = WithOpenFile()
            with_open_file.open_file(file_obj=file_obj, sftp=sftp,
                                     remote_path=remote_path, group_name=md5_group)
        except Exception as e:
            self.file_record(remote_path=remote_path, hostname=host_obj.hostname, username=username, code=2)
            error_msg = f"主机 {host_id} 分发失败: {str(e)}"
            with lock:
                errors.append(error_msg)
            logger.error(error_msg)
        finally:
            if not error_msg:
                self.file_record(remote_path=remote_path, hostname=host_obj.hostname, username=username, code=1)
            if sftp:
                sftp.close()
            if ssh_client:
                ssh_client.close()

    def file_record(self, remote_path, username='', hostname='', code=0, dest_hostname='', src_hostname=''):
        if dest_hostname and src_hostname:
            hosts = f"{src_hostname} => {dest_hostname}"
        else:
            hosts = hostname
        FileDistribution.objects.create(path=remote_path, hostname=hosts, user=username, status=code)
        file_record_obj = FileDistribution.objects.filter(user=username)
        if file_record_obj.count() > 100:
            file_record_obj.order_by('id').first().delete()


class TerminalFileUploadView(APIView):
    """terminal页面文件上传功能"""

    @check_permission(perms_list=['host:list'])  # 直接应用装饰器，无需method_decorator
    def dispatch(self, request, *args, **kwargs):
        return super().dispatch(request, *args, **kwargs)

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.__encryption_pass = Password()

    def post(self, request):
        """文件上传"""
        data = request.POST.dict()
        host_id = data.get("host_id")
        host_obj = HostList.objects.filter(id=host_id).get()
        if host_obj.server_type == 1:
            path = data.get("path").rstrip('/')
            if not path.startswith('/tmp'):
                return JsonResponse({"code": 4001, "msg": "没有目录权限"})
        else:
            path = data.get("path").rstrip('\\')
            if not path.startswith(fr'C:\Users\{host_obj.user}\Downloads'):
                return JsonResponse({"code": 4001, "msg": "没有目录权限"})
        try:
            # 建立SFTP连接
            get_sftp = TerminalFileView()
            sftp, ssh_client = get_sftp.get_sftp_connection(host_id)
            return self.__handle_upload(request, path, sftp, ssh_client, host_obj)
        except Exception as e:
            logger.error(f"文件操作失败: {str(e)}")
            return JsonResponse({"code": 500, "msg": f"文件操作失败：{str(e)}"})
        finally:
            sftp.close()
            ssh_client.close()

    def __handle_upload(self, request, path, sftp, ssh_client, host_obj):
        """处理文件上传"""
        file_obj = request.FILES['files']
        # 100MB限制
        if file_obj.size > 1024 * 1024 * 1000:
            return JsonResponse({"code": 400, "msg": "文件大小超过限制"})
        try:
            if host_obj.server_type == 1:
                remote_path = path + '/' + file_obj.name
            else:
                remote_path = path + '\\' + file_obj.name
            # 初始化文件传输类
            with_open_file = WithOpenFile()
            with_open_file.open_file(file_obj=file_obj, sftp=sftp, remote_path=remote_path, group_name="terminal")

            return JsonResponse({"code": 200, "msg": "上传成功"})
        except Exception as e:
            raise e
        finally:
            sftp.close()
            ssh_client.close()


class TerminalFileView(APIView):
    """terminal页面文件管理功能"""

    @check_permission(perms_list=['host:list'])  # 直接应用装饰器，无需method_decorator
    def dispatch(self, request, *args, **kwargs):
        return super().dispatch(request, *args, **kwargs)

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.__encryption_pass = Password()

    def get_sftp_connection(self, host_id):
        """建立SFTP连接"""
        try:
            host_obj = HostList.objects.filter(id=int(host_id)).first()
            if not host_obj:
                raise Exception("主机不存在")

            # 解密密码
            password = self.__encryption_pass.decrypt(host_obj.password)
            print(host_obj.connect_ip)
            # 创建SSH连接
            ssh_client = SSHClient()
            ssh_client.set_missing_host_key_policy(AutoAddPolicy())
            ssh_client.connect(
                hostname=host_obj.connect_ip,
                port=int(host_obj.port),
                username=host_obj.user,
                password=password,
                timeout=10
            )
            # 创建SFTP客户端
            return [ssh_client.open_sftp(), ssh_client]
        except Exception as e:
            logger.error(f"SFTP连接失败: {str(e)}")
            return False

    def post(self, request):
        """获取文件列表"""
        data = json.loads(request.body.decode("utf-8"))

        host_id = data.get("host_id")
        host_obj = HostList.objects.filter(id=int(host_id)).get()
        if host_obj.server_type == 2:
            path = data.get("path").rstrip('\\')
            if not path.startswith(fr'C:\Users\{host_obj.user}\Downloads'):
                return JsonResponse({"code": 4001, "msg": "没有目录权限"})
            # 建立SFTP连接
            sftp, ssh_client = self.get_sftp_connection(host_id)
            if not sftp:
                return JsonResponse({"code": 500, "msg": "无法连接服务器"})
            return self.__handle_list(path, sftp, ssh_client)

        path = data.get("path").rstrip('/')
        if not path.startswith('/tmp'):
            return JsonResponse({"code": 4001, "msg": "没有目录权限"})

        try:
            # 建立SFTP连接
            sftp, ssh_client = self.get_sftp_connection(host_id)
            return self.__handle_list(path, sftp, ssh_client)
        except Exception as e:
            logger.error(f"文件操作失败: {str(e)}")
            return JsonResponse({"code": 500, "msg": f"文件操作失败：{str(e)}"})
        finally:
            sftp.close()
            ssh_client.close()

    def __handle_list(self, path, sftp, ssh_client):
        """处理文件列表请求"""
        try:
            files = []
            dirs = []
            sorted_all_files = sorted(sftp.listdir_attr(path), key=lambda x: x.st_mtime, reverse=True)
            for item in sorted_all_files:
                if item.filename in ['.', '..']:
                    continue
                file_info = {
                    "name": item.filename,
                    "is_dir": bool(item.st_mode & 0o40000),
                    "size": item.st_size,
                    "mtime": time.strftime("%Y-%m-%d %H:%M:%S",
                                           time.localtime(item.st_mtime))
                }
                if bool(item.st_mode & 0o40000):
                    dirs.append(file_info)
                else:
                    files.append(file_info)
            files += dirs
            return JsonResponse({"code": 200, "data": files})

        except FileNotFoundError:
            return JsonResponse({"code": 404, "msg": "路径不存在，返回/tmp"})
        except Exception as e:
            raise e
        finally:
            sftp.close()
            ssh_client.close()

    async def file_iterator(self, file_obj, chunk_size=8192):
        """异步文件读取迭代器"""
        while True:
            chunk = file_obj.read(chunk_size)
            if not chunk:
                break
            yield chunk
            # 让出控制权
            await asyncio.sleep(0)

    def get(self, request):
        """处理文件下载"""
        host_id = request.GET.get("host_id")
        path = request.GET.get("path")
        filename = request.GET.get("filename")

        path = urllib.parse.unquote(path)
        filename = urllib.parse.unquote(filename)
        file_path = f"{path}/{filename}"

        try:
            if not all([host_id, path, filename]):
                return JsonResponse({"code": 400, "msg": "参数不完整"})

            # 获取SFTP连接
            sftp, ssh_client = self.get_sftp_connection(host_id)
            if not sftp:
                return JsonResponse({"code": 500, "msg": "无法连接服务器"})

            # 打开文件并保持连接
            file_obj = sftp.open(file_path, 'rb')

            def client_close():
                sftp.close()
                file_obj.close()
                ssh_client.close()

            response = FileResponseLocal(
                client_close,
                self.file_iterator(file_obj),
                as_attachment=True,
                filename=filename
            )
            return response

        except FileNotFoundError:
            return JsonResponse({"code": 404, "msg": "文件不存在"})
        except Exception as e:
            logger.error(f"下载失败: {str(e)}")
            return JsonResponse({"code": 500, "msg": f"下载失败：{str(e)}"})


class GetRelatedGroups:
    """获取所有相关的组ID"""

    def get_related_groups(self, group_id=None):
        group_list = []
        group_obj = GroupList.objects.filter(parent_id=group_id).values('id')
        for item in group_obj:
            group_list.append(item['id'])
        child_group = GroupList.objects.filter(parent_id__in=group_list).values('id')
        for item in child_group:
            group_list.append(item['id'])
        group_list.insert(0, group_id)
        return group_list


class AuthRoleGroupView:
    """获取用户角色权限，并获取角色的主机分组"""

    def auth(self, request):
        username = request.session.get('username')
        role_list = SysUserRole.objects.distinct().filter(user__username=username).values_list('role_id')
        groups_obj = GroupList.objects.distinct().filter(sysrolegrouplist__role_id__in=role_list).values()
        return groups_obj


class HostListView(APIView):
    """获取主机列表"""

    @check_permission(perms_list=['host:list', 'job:list'])  # 直接应用装饰器，无需method_decorator
    def dispatch(self, request, *args, **kwargs):
        return super().dispatch(request, *args, **kwargs)

    def post(self, request):
        data = json.loads(request.body.decode("utf-8"))
        group_id = data["group_id"]
        query = data["query"].strip()
        # 当前页
        page_num = data["pageNum"]
        # 每页大小
        page_size = data["pageSize"]

        query_criteria = Q()
        if query:
            query_criteria &= Q(hostname__contains=query) | Q(public_ip_addr=query) | Q(private_ip_addr=query)

        if group_id:
            get_related_groups = GetRelatedGroups()
            group_list = get_related_groups.get_related_groups(group_id)
            query_criteria &= Q(grouphost__group_id__in=group_list)
        else:
            auth_role_groups = AuthRoleGroupView()
            groups_obj = auth_role_groups.auth(request).values_list('id')
            query_criteria &= Q(grouphost__group_id__in=groups_obj)
        try:
            host_queryset = (HostList.objects.distinct().filter(query_criteria).order_by("sort_num")
                             .values('id', 'hostname', 'public_ip_addr', 'private_ip_addr', 'create_time',
                                     'update_time', 'config_info', 'port', 'user', 'status', 'remark', 'connect_ip',
                                     'server_type'))
            # 使用PageData 类，获取分页数据
            host_list_obj = PageData(host_queryset, page_size, page_num)
            host_list = host_list_obj.get_search()
            for item in host_list:
                try:
                    item['create_time'] = item['create_time'].strftime('%Y-%m-%d %H:%M:%S')
                    item['update_time'] = item['update_time'].strftime('%Y-%m-%d %H:%M:%S')
                except AttributeError as e:
                    logger.error(str(e))
                group_id_set = set()
                group_obj = GroupHost.objects.filter(host_id=item['id']).values('group_id')
                for obj in group_obj:
                    group_id_set.add(obj['group_id'])
                item['groups'] = list(group_id_set)
            total = host_queryset.count()
        except Exception as e:
            logger.error(e)
            return JsonResponse({"code": 500, "msg": f"主机列表获取失败：{str(e)}"})
        return JsonResponse({"code": 200, "total": total, "msg": host_list})


def get_host_msg(connect_ip, port, user, password, server_type):
    ssh_client = LocalSSHClient(ip_addr=connect_ip, port=int(port), username=user, password=password.strip())
    if server_type == 1:
        msg_dict = ssh_client.get_all_msg()
    else:
        msg_dict = ssh_client.get_win_msg()
    os_release = msg_dict['os_release']
    os_conf = msg_dict['os_conf']

    host_info = f"{os_release} {os_conf}"
    encrypt_password = Password().encrypt(password)
    # 获取主机公网IP
    public_ip_addr = msg_dict['public_ip']
    # 获取主机内网IP
    private_ip_addr = msg_dict['private_ip']
    return host_info, encrypt_password, public_ip_addr, private_ip_addr


class HostsBatchImportView(APIView):
    """主机批量导入"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.__encryption_pass = Password()

    @check_permission(perms_list=['host:list'])  # 直接应用装饰器，无需method_decorator
    def dispatch(self, request, *args, **kwargs):
        return super().dispatch(request, *args, **kwargs)

    def post(self, request):
        file_obj = request.FILES['files']
        group_id = request.POST.get('id')
        excel_obj = pandas.read_excel(file_obj, engine='openpyxl', usecols=range(9))
        group_obj = GroupList.objects.filter(id=group_id).get()
        if not group_obj:
            return JsonResponse({"code": 500, "msg": f"主机批量导入失败：主机分组不存在"})
        try:
            for index, row in excel_obj.iterrows():
                try:
                    for k, v in row.items():
                        if v == 'nan':
                            row[k] = None
                    if str(row['rdp_port']) != 'nan':
                        row['rdp_port'] = int(row['rdp_port'])
                    else:
                        row['rdp_port'] = None
                    HostList.objects.create(hostname=row['host_name'], connect_ip=row['ip_addr'], port=row['ssh_port'],
                                            user=row['user_name'],
                                            password=self.__encryption_pass.encrypt(str(row['password'])),
                                            sort_num=row['sort_num'], server_type=row['server_type'],
                                            rdp_port=row['rdp_port'], remark=row['remark'], status=1)

                    if group_obj:
                        GroupHost.objects.create(host_id=HostList.objects.filter(hostname=row['host_name']).get().id,
                                                 group_id=group_id)
                except IntegrityError as e:
                    logger.error(e)
                    continue
        except Exception as e:
            logger.error(f"主机批量导入失败：{e}")
            return JsonResponse({"code": 500, "msg": f"主机批量导入失败：{str(e)}"})
        return JsonResponse({"code": 200})


class VerifyHostView(APIView):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.__encryption_pass = Password()

    @check_permission(perms_list=['host:list'])  # 直接应用装饰器，无需method_decorator
    def dispatch(self, request, *args, **kwargs):
        return super().dispatch(request, *args, **kwargs)

    def __process_host(self, item):
        close_old_connections()  # 确保每个线程使用新的连接

        try:
            # 解密密码
            password = self.__encryption_pass.decrypt(item['password'])

            # 获取主机信息
            host_info, _, public_ip_addr, private_ip_addr = get_host_msg(
                item['connect_ip'],
                item['port'],
                item['user'],
                password,
                item['server_type']
            )

            # 更新成功状态
            HostList.objects.filter(id=item['id']).update(
                public_ip_addr=public_ip_addr,
                private_ip_addr=private_ip_addr,
                config_info=host_info,
                status=2
            )
            return None
        except Exception as e:
            # 更新失败状态
            HostList.objects.filter(id=item['id']).update(status=0)
            logger.error(f"验证主机失败：{item['hostname']} {str(e)}")
            return f"主机_{item['hostname']}: {str(e)}"
        finally:
            close_old_connections()  # 清理数据库连接

    def post(self, request):
        data = json.loads(request.body.decode("utf-8"))
        role_list = request.session.get("role_list")
        group_list = GroupList.objects.filter(sysrolegrouplist__role__in=role_list).values('id')
        query = Q(grouphost__group_id__in=group_list)

        if data.get('all', 0) != 1:  # 更安全的get方法
            query &= Q(status__in=[0, 1])

        # 强制转换为列表避免数据库游标问题
        host_obj = list(HostList.objects.filter(query).values(
            'id', 'password', 'connect_ip', 'port', 'user', 'hostname', 'server_type'
        ))

        errors = []

        # 使用线程池并发执行
        from concurrent.futures import ThreadPoolExecutor

        with ThreadPoolExecutor(max_workers=10) as executor:  # 根据需要调整线程数
            # 提交所有任务
            future_results = [executor.submit(self.__process_host, item) for item in host_obj]

            # 收集结果
            for future in future_results:
                error = future.result()
                if error:
                    errors.append(error)

        if errors:
            return JsonResponse({
                "code": 500,
                "msg": f"批量验证完成，其中{len(errors)}台失败：{', '.join(errors)}"
            })
        return JsonResponse({"code": 200, "msg": "所有主机验证成功"})


class HostActionView(APIView):
    """主机操作"""

    @check_permission(perms_list=['host:list'])  # 直接应用装饰器，无需method_decorator
    def dispatch(self, request, *args, **kwargs):
        return super().dispatch(request, *args, **kwargs)

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.__encryption_pass = Password()

    def get(self, request):
        host_id = request.GET.get("id")
        host_obj = (HostList.objects.filter(id=host_id)
                    .values('id', 'hostname', 'public_ip_addr', 'private_ip_addr', 'create_time',
                            'config_info', 'port', 'rdp_port', 'user', 'status', 'remark', 'connect_ip', 'sort_num',
                            'server_type')
                    .first())
        group_id_set = set()
        group_obj = GroupHost.objects.filter(host_id=host_obj['id']).values('group_id')
        for obj in group_obj:
            group_id_set.add(obj['group_id'])
        host_obj['groups'] = list(group_id_set)
        return JsonResponse({"code": 200, "msg": host_obj})

    def post(self, request):
        """添加和修改主机"""
        data = json.loads(request.body.decode("utf-8"))
        host_id = data['id']
        hostname = data['hostname']
        user = data['user']
        port = data['port']
        rdp_port = data.get('rdp_port')
        if not data.get('rdp_port'):
            rdp_port = None
        connect_ip = data['connect_ip']
        password = data.get('password')
        remark = data['remark']
        groups = data['groups']
        create_time = data['create_time']
        sort_num = data['sort_num']
        server_type = data['server_type']
        host_obj = None
        if len(groups) < 1:
            return JsonResponse({"code": 500, "msg": "请至少选择一个分组"})
        try:

            if host_id == -1:
                host_info, encrypt_password, public_ip_addr, private_ip_addr = get_host_msg(connect_ip, port, user,
                                                                                            password, server_type)
                host_obj = HostList(hostname=hostname, user=user, config_info=host_info, status=2,
                                    port=port, rdp_port=rdp_port, public_ip_addr=public_ip_addr,
                                    private_ip_addr=private_ip_addr, connect_ip=connect_ip,
                                    password=encrypt_password, remark=remark, sort_num=sort_num,
                                    server_type=server_type)
            else:
                filter_q = Q()
                filter_q &= (Q(sort_num=sort_num) & Q(id=host_id) & Q(hostname=hostname) & Q(user=user)
                             & Q(connect_ip=connect_ip) & Q(port=port) & Q(remark=remark)
                             & Q(grouphost__group_id__in=groups) & Q(server_type=server_type))

                decrypt_password = ''
                if not password:
                    tmp_pass = HostList.objects.filter(id=host_id).values('password').first()['password']
                    decrypt_password = self.__encryption_pass.decrypt(tmp_pass)

                if password:
                    filter_q &= Q(password=self.__encryption_pass.encrypt(password))
                    decrypt_password = password

                if rdp_port:
                    filter_q &= Q(rdp_port=rdp_port)

                # 判断前端传的数据是否有改动，没有改动，则直接返回200状态码
                exits_host_obj = HostList.objects.filter(filter_q)
                exits_group_obj = GroupHost.objects
                exits_group_list = exits_group_obj.filter(host_id=host_id).values('group_id')
                group_list = []
                for item in exits_group_list:
                    group_list.append(item['group_id'])
                if exits_host_obj.exists():
                    if sorted(group_list) == sorted(groups):
                        return JsonResponse({"code": 200})

                host_info, password, public_ip_addr, private_ip_addr = get_host_msg(connect_ip, port, user,
                                                                                    decrypt_password,
                                                                                    HostList.objects.filter(
                                                                                        id=host_id).get().server_type)
                host_obj = HostList(id=host_id, hostname=hostname, user=user, config_info=host_info, status=2,
                                    port=port, rdp_port=rdp_port, public_ip_addr=public_ip_addr,
                                    private_ip_addr=private_ip_addr, connect_ip=connect_ip,
                                    create_time=create_time, password=password, remark=remark,
                                    sort_num=sort_num, server_type=server_type)

                GroupHost.objects.filter(host_id=host_id).delete()

            host_obj.save()
            host_id = HostList.objects.filter(hostname=hostname).values('id').first()['id']
            for item in groups:
                GroupHost.objects.create(host_id=host_id, group_id=item)

        except Exception as e:
            logger.error(e)
            return JsonResponse({"code": 500, "msg": f"主机操作失败：{str(e)}"})
        return JsonResponse({"code": 200})

    def delete(self, request):
        """删除主机数据"""
        data = json.loads(request.body.decode("utf-8"))
        host_id = data['id']
        try:
            # host_obj = HostList.objects.filter(id=host_id).get()
            # FileDistribution.objects.filter(hostname=host_obj.hostname).delete()  # 删除文件分发记录
            # CmdRecord.objects.filter(hostname=host_obj.hostname).delete()  # 删除命令执行记录
            GroupHost.objects.filter(host_id=host_id).delete()
            HostList.objects.filter(id=host_id).delete()
            return JsonResponse({"code": 200})
        except Exception as e:
            logger.error(e)
            return JsonResponse({"code": 500, "msg": f"主机操作失败：{str(e)}"})


class GroupListView(APIView):
    """获取服务器分组列表"""

    @check_permission(perms_list=['host:list', 'job:list'])  # 直接应用装饰器，无需method_decorator
    def dispatch(self, request, *args, **kwargs):
        return super().dispatch(request, *args, **kwargs)

    def build_tree_menu(self, group_list, status):
        result_group_list = []
        get_related_groups = GetRelatedGroups()
        num = 0
        for group in group_list:
            tmp_list = []
            # 寻找子节点
            for item in group_list:
                if item["parent_id"] == group["id"]:
                    item_id = get_related_groups.get_related_groups(item["id"])
                    item["count"] = (HostList.objects.distinct()
                                     .filter(grouphost__group_id__in=item_id).count())
                    tmp_list.append(item)
            group["children"] = tmp_list

            # 给前端增加index
            num += 1
            group['index'] = 'group_' + str(num)
            group['type'] = 'node'

            if status == 1:
                group['label'] = group['name']
                del group['name']
                host_obj = (HostList.objects.filter(grouphost__group_id=group['id']).order_by('sort_num')
                            .values('id', 'hostname', 'public_ip_addr', 'private_ip_addr', 'connect_ip', 'config_info',
                                    'port', 'rdp_port', 'user', 'status', 'remark', 'server_type'))
                for item in host_obj:
                    item['index'] = 'host_' + item['hostname']
                    item['type'] = 'host'
                    item['label'] = item['hostname']
                    del item['hostname']
                    group['children'].append(item)

            # 判断父节点，添加到集合
            if group["parent_id"] == 0:
                group_id = get_related_groups.get_related_groups(group["id"])
                group["count"] = (HostList.objects.distinct()
                                  .filter(grouphost__group_id__in=group_id).count())
                result_group_list.append(group)

        return result_group_list

    def get(self, request):
        try:
            status = request.GET.get('status')
            if status:
                status = int(status)
            # 判断是否是 角色管理/主机权限/主机分组列表 传来的请求
            if status == 2:
                group_queryset = GroupList.objects.order_by('id').values()
            else:
                auth_role_groups = AuthRoleGroupView()
                groups_obj = auth_role_groups.auth(request)
                group_queryset = groups_obj.order_by('id').values()

            # 构造菜单树
            group_list = self.build_tree_menu(group_list=list(group_queryset), status=status)
        except Exception as e:
            logger.error(e)
            return JsonResponse({"code": 500, "msg": "分组列表树构造失败"})
        return JsonResponse({"code": 200, "msg": group_list})


class GroupActionView(APIView):
    """分组操作"""

    @check_permission(perms_list=['host:list'])  # 直接应用装饰器，无需method_decorator
    def dispatch(self, request, *args, **kwargs):
        return super().dispatch(request, *args, **kwargs)

    def post(self, request):
        """添加和修改分组"""
        data = json.loads(request.body.decode("utf-8"))
        try:
            parent_id = data['parent_id']
            if data['id'] == -1:  # 添加
                group_list_obj = GroupList(name=data['name'], parent_id=parent_id)
                group_list_obj.save()

                # 给分组增加层级数字
                if parent_id:
                    layered_id = 1
                    group_obj = GroupList.objects
                    one_parent = group_obj.filter(id=parent_id)
                    if one_parent:
                        layered_id += 1
                        two_parent_id = group_obj.filter(id=one_parent.values('parent_id').first()['parent_id'])
                        if two_parent_id:
                            layered_id += 1

                    group_list_obj.layered_id = layered_id
                    group_list_obj.save()
                else:
                    group_list_obj.layered_id = 1
                    group_list_obj.save()
            else:  # 修改
                group_id = data['id']
                group = GroupList.objects.get(id=group_id)
                original_parent_id = group.parent_id  # 保存原父组ID用于后续判断
                new_parent_id = parent_id

                # 更新分组基本信息
                group.name = data['name']
                group.parent_id = new_parent_id

                # 计算新的层级ID
                new_layered_id = 1
                if new_parent_id:  # 新父组存在
                    new_parent = GroupList.objects.get(id=new_parent_id)
                    new_layered_id = new_parent.layered_id + 1
                group.layered_id = new_layered_id

                # 保存修改
                group.save()

                # 如果父组发生变化，递归更新所有子组的层级
                if original_parent_id != new_parent_id:
                    def update_children_layered(current_group):
                        children = GroupList.objects.filter(parent_id=current_group.id)
                        for child in children:
                            child.layered_id = current_group.layered_id + 1
                            child.save()
                            update_children_layered(child)  # 递归更新子组

                    update_children_layered(group)
        except Exception as e:
            logger.error(e)
            return JsonResponse({"code": 500, "msg": f"操作分组失败：{str(e)}"})
        return JsonResponse({"code": 200})

    def get(self, request):
        """查询新增的group数据"""
        group_id = request.GET.get("id")
        group_obj = GroupList.objects.filter(id=group_id).values().first()
        return JsonResponse({"code": 200, "msg": group_obj})

    def delete(self, request):
        """删除group数据"""
        group_id = json.loads(request.body.decode("utf-8"))
        try:
            if GroupList.objects.filter(parent_id=group_id).count() > 0:
                return JsonResponse({"code": 500, "msg": "请先删除子分组"})
            elif HostList.objects.filter(grouphost__group_id=group_id).count() > 0:
                return JsonResponse({"code": 500, "msg": "该分组中还有主机信息，请先把主机移动到其他分组"})
            else:
                # 删除组和角色关联数据
                SysRoleGroupList.objects.filter(group_id=group_id).delete()
                # 删除组
                GroupList.objects.filter(id=group_id).delete()
        except Exception as e:
            logger.error(e)
            return JsonResponse({"code": 500, "msg": f"删除失败：{str(e)}"})
        return JsonResponse({"code": 200})


class JobView(APIView):
    """命令批量执行"""

    @check_permission(perms_list=['job:list'])  # 直接应用装饰器，无需method_decorator
    def dispatch(self, request, *args, **kwargs):
        return super().dispatch(request, *args, **kwargs)

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.__encryption_pass = Password()

    def post(self, request):
        data = json.loads(request.body.decode("utf-8"))
        host_list = data.get('host_list')
        cmd = data.get('cmd')
        if cmd.isspace() or cmd == "":
            return JsonResponse({"code": 500, "msg": "请输入需要执行的命令！"})

        processes = []
        error_queue = multiprocessing.Queue()
        lock = multiprocessing.Lock()
        host_queue = multiprocessing.Queue()

        try:
            for host_id in host_list:
                host_obj = HostList.objects.filter(id=host_id).first()
                password = self.__encryption_pass.decrypt(host_obj.password)
                ip_addr = host_obj.connect_ip
                port = int(host_obj.port)
                username = host_obj.user
                hostname = host_obj.hostname
                server_type = host_obj.server_type

                process = multiprocessing.Process(
                    target=self.__cmd_process,
                    args=(
                        request, server_type, host_id, hostname, username, port, ip_addr, password, cmd, lock,
                        error_queue,
                        host_queue)
                )
                processes.append(process)
                process.start()

            for process in processes:
                process.join()

            user = request.session.get('username')
            cmd_obj = CmdRecord.objects.filter(user=user)

            if cmd_obj.count() > 100:
                num = cmd_obj.order_by('-id').first().id - 100
                cmd_obj.filter(id__lt=num).delete()

            errors = []
            while not error_queue.empty():
                errors.append(error_queue.get())
            err_host_id = []
            while not host_queue.empty():
                err_host_id.append(host_queue.get())

            if errors:
                return JsonResponse({
                    "code": 207,
                    "msg": f"命令批量执行未完成，失败{len(errors)}台：",
                    "errors": errors,
                    "err_host_id": err_host_id,
                })
        except Exception as e:
            logger.error(e)
        return JsonResponse({"code": 200, "msg": "命令批量执行成功"})

    def __cmd_process(self, request, server_type, host_id, hostname, username, port, ip_addr, password, cmd, lock,
                      error_queue,
                      host_queue):
        user = request.session.get('username')
        try:
            logger.info(f"正在连接主机 {hostname} ({ip_addr})...")
            client = LocalSSHClient(ip_addr=ip_addr, port=port, username=username, password=password)
            logger.info(f"连接成功，执行命令: {cmd}")

            client.execute_command_ws(server_type, cmd, host_id, hostname, user, username)
            logger.info(f"主机 {hostname} 命令执行完成")
        except Exception as e:
            error_msg = f"主机：{hostname}，错误: {str(e)}"
            logger.error(error_msg, exc_info=True)
            with lock:
                host_queue.put(host_id)
                error_queue.put(error_msg)

    def get(self, request):
        try:
            user = request.session.get('username')
            cmd_list = CmdRecord.objects.filter(user=user).order_by('-id').values()
            for item in cmd_list:
                try:
                    item['execution_time'] = item['execution_time'].strftime('%Y-%m-%d %H:%M:%S')
                except AttributeError as e:
                    logger.error(str(e))
            return JsonResponse({"code": 200, "msg": list(cmd_list), "total": cmd_list.count()})
        except Exception as e:
            return JsonResponse({"code": 500, "msg": str(e)})
